package org.yonggan.util;

import com.alibaba.fastjson.JSONObject;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import org.yonggan.conf.ConfigurationManager;
import org.yonggan.constant.Constants;
import org.yonggan.mock.MockData;

/**
 * 公共
 */
public class SparkUtils {

    /**
     * 设置本地的master
     * @param conf
     */
    public static void setMaster(SparkConf conf) {
        if (ConfigurationManager.getBoolean(Constants.SPARK_LOCAL))
            conf.setMaster("local[*]");
    }

    /**
     * 获取 SQLContext
     *
     * @param sc
     * @return
     */
    public static SQLContext getSQLContext(SparkContext sc) {

        Boolean bool = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);

        if (bool) {
            return new SQLContext(sc); // 本地环境
        } else {
            return new HiveContext(sc); // 生产环境
        }
    }


    /**
     * 生成模拟数据
     * 如spark.local =  true 则会生成数据
     *
     * @param sc
     * @param sqlContext
     */
    public static void makeMock(JavaSparkContext sc, SQLContext sqlContext) {

        Boolean bool = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
        if (bool) {
            MockData.mock(sc, sqlContext);
        }
    }


    /**
     * 获取指定日期范围内的用户访问行为数据RDD
     *
     * @param sqlContext SQLContext
     * @param taskParam  任务参数
     * @return 行为数据RDD
     */
    public static JavaRDD<Row> getActionRDDByDateRange(
            SQLContext sqlContext, JSONObject taskParam) {
        String startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE);
        String endDate = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE);

        String sql =
                "select * "
                        + "from user_visit_action "
                        + "where date>='" + startDate + "' "
                        + "and date<='" + endDate + "'";
//				+ "and session_id not in('','','')"

        DataFrame actionDF = sqlContext.sql(sql);

        /**
         * 这里就很有可能发生上面说的问题
         * 比如说，Spark SQl默认就给第一个stage设置了20个task，但是根据你的数据量以及算法的复杂度
         * 实际上，你需要1000个task去并行执行
         *
         * 所以说，在这里，就可以对Spark SQL刚刚查询出来的RDD执行repartition重分区操作
         */

//		return actionDF.javaRDD().repartition(1000);

        return actionDF.javaRDD();
    }


}
